feat: implment bulk radius api (CM-1328) - #4390

backend/src/api/public/v1/packages/submitBlastRadiusJobBatch.ts:35

• getPackagesTemporalClient() establishes the Temporal connection and can reject when Temporal is unreachable (backend/src/db/packagesTemporal.ts:18-42). Because that happens before the per-job try blocks, the entire request fails instead of returning a 202 with failed entries. Move client acquisition into submitOneJob's failure boundary (the getter already caches the connection promise) so this runtime failure follows the documented per-job semantics.

  const packagesTemporal = await getPackagesTemporalClient()

backend/src/api/public/v1/packages/submitBlastRadiusJobBatch.ts:92

• If failAnalysis rejects, submitOneJob still rejects and Promise.all fails the whole batch. This directly defeats the stated isolation for createAnalysis failures (a database error may also make this follow-up write fail). Guard the failure-recording write separately, log that persistence failure, and still return this job's failed entry if the endpoint contract must always isolate per-job runtime failures.

    await blastRadiusDal.failAnalysis(qx, analysisInput, errorMessage)

backend/src/api/public/v1/packages/submitBlastRadiusJobBatch.ts:35

• getPackagesTemporalClient() establishes the connection and can reject when Temporal is unreachable. Because it runs before jobs.map(...), a cold-start outage fails the entire HTTP request instead of returning one failed entry per job with 202 as the endpoint contract promises. Move client acquisition into the per-job try path (the cached _init promise still prevents 20 independent successful connections).

  const packagesTemporal = await getPackagesTemporalClient()

backend/src/api/public/v1/packages/submitBlastRadiusJobBatch.ts:92

• This awaited cleanup can itself reject, which escapes submitOneJob and makes Promise.all reject the whole batch. In particular, the newly handled createAnalysis DB failure is likely to make this second DB write fail too, so the advertised per-job isolation is not guaranteed. Handle/log failure persistence separately so submitOneJob still resolves to its failed entry.

    await blastRadiusDal.failAnalysis(qx, analysisInput, errorMessage)

backend/src/api/public/v1/packages/getBlastRadiusJobBatch.ts:51

• Valid UUID input is case-insensitive, while PostgreSQL serializes UUID columns in lowercase. An uppercase analysisId therefore passes z.uuid() but misses these case-sensitive maps, returning found: false (or empty verdict/count data) for an existing analysis. Normalize lookup keys while preserving requestedAnalysisId for the response echo.

  const results: BlastRadiusAnalysisBulkEntry[] = pagedAnalysisIds.map((requestedAnalysisId) => {
    const analysis = analysisById.get(requestedAnalysisId)

backend/src/api/public/v1/packages/submitBlastRadiusJobBatch.ts:35

• Temporal client initialization happens before the per-job boundary. getPackagesTemporalClient() performs Connection.connect() and can reject when Temporal is unreachable, so this path returns a batch-wide 5xx before creating any analysis rows instead of the documented 202 with failed entries. Acquire/await the client inside submitOneJob's try (the cached initializer will still share one connection attempt) so this failure is mapped per job.

  const packagesTemporal = await getPackagesTemporalClient()

backend/src/api/public/v1/packages/submitBlastRadiusJobBatch.ts:92

• This awaited recovery write can itself reject, causing submitOneJob to reject and Promise.all to fail the entire request. That breaks the endpoint's stated per-job isolation, especially for the createAnalysis database-error path this catch is intended to handle. Handle failure persistence separately (with appropriate logging/response semantics) so the helper cannot unexpectedly reject.

    await blastRadiusDal.failAnalysis(qx, analysisInput, errorMessage)

backend/src/api/public/v1/akrites-external/index.ts:31

• This also raises the existing single-job endpoint's default limit from 5 to 50 requests/hour. Combined with 20 jobs per batch, the default now permits up to 1,000 workflow/LLM starts per hour, while the PR describes retaining the strict limiter and does not document this separate 10× relaxation. Keep the previous default unless this capacity and cost increase was explicitly approved.

      : 50,

backend/src/api/public/v1/packages/submitBlastRadiusJobBatch.ts:38

• Temporal connection setup happens before the per-job failure boundary. getPackagesTemporalClient() calls Connection.connect, so if Temporal is unreachable on a cold connection this rejects the entire request with a 500 and no per-job results, contrary to the documented isolation semantics. Acquire the client inside each job's guarded path (the shared initializer will still deduplicate the connection), or map an initialization failure to failed entries for every job.

  const packagesTemporal = await getPackagesTemporalClient()

  const results: BlastRadiusJobEntry[] = await Promise.all(
    jobs.map((body) => submitOneJob(qx, packagesTemporal, body)),

backend/src/api/public/v1/packages/submitBlastRadiusJobBatch.ts:92

• A rejection from failAnalysis still escapes submitOneJob and rejects the outer Promise.all, so the promised per-job isolation is not guaranteed. This is especially likely after createAnalysis failed because the same database problem can make this recovery write fail. Handle and log the secondary persistence failure while still returning the failed entry, or use an all-settled aggregation that explicitly maps every rejection.

    await blastRadiusDal.failAnalysis(qx, analysisInput, errorMessage)

backend/src/api/public/v1/akrites-external/index.ts:31

• Raising this request-based limit from 5 to 50 lets one client enqueue up to 1,000 workflows per hour through 20-item batches, versus five workflows per hour before this PR. That materially increases Temporal/LLM load and cost while the PR description explicitly notes that workflow-weighted limiting was not implemented. Keep the existing default unless the limiter is changed to account for batch size.

      : 50,

backend/src/api/public/v1/akrites-external/openapi.yaml:462

• The response can contain status: failed precisely because a job was not successfully submitted (for example, workflow.start failed), so saying every job is submitted is inaccurate for API consumers. Document that every requested job receives an entry and that failed entries may represent submission failures.

      description: >
        Plain array in request order, one entry per submitted job — unlike the
        read batches there is no found/not-found case, every job is submitted.

backend/src/api/public/v1/packages/submitBlastRadiusJobBatch.ts:35

• getPackagesTemporalClient() establishes the Temporal connection before entering any per-job try/catch. On a cold request while Temporal is unreachable, this rejects the whole handler with a 500, rather than returning a 202 with one failed result per job as the batch contract promises. Move client acquisition into the per-job failure boundary (while preserving the shared cached connection) or explicitly convert acquisition failure into per-job failed entries.

  const packagesTemporal = await getPackagesTemporalClient()

backend/src/api/public/v1/packages/submitBlastRadiusJobBatch.ts:92

• This recovery write can itself reject. If the database remains unavailable after createAnalysis fails, or if marking a failed workflow start fails, submitOneJob rejects and the surrounding Promise.all aborts the entire response, contradicting the documented per-job isolation. Handle and log a failAnalysis failure without allowing it to reject the batch, or use an all-settled aggregation that still produces a result for every input.

    const errorMessage = err instanceof Error ? err.message : String(err)
    await blastRadiusDal.failAnalysis(qx, analysisInput, errorMessage)

backend/src/api/public/v1/packages/getBlastRadiusJobBatch.ts:51

• UUID text is case-insensitive, but PostgreSQL returns UUID columns in canonical lowercase while z.uuid() accepts uppercase hex. An uppercase analysisId therefore queries successfully but misses these case-sensitive Maps, returning found: false (and would also miss verdict/count buckets). Normalize only the lookup key while preserving requestedAnalysisId in the echoed response.

  const results: BlastRadiusAnalysisBulkEntry[] = pagedAnalysisIds.map((requestedAnalysisId) => {
    const analysis = analysisById.get(requestedAnalysisId)

services/apps/packages_worker/src/blast-radius/workflows.ts:41

• The report activity only calls heartbeat() after runReportStage has completed. Adding a one-minute heartbeat timeout therefore makes any otherwise-valid report taking 1–2 minutes time out before its first heartbeat, reducing the effective limit below the existing two-minute start-to-close timeout. Remove this heartbeat timeout, or pass a heartbeat callback into the report stage and invoke it during the work.

  heartbeatTimeout: '1 minute',

backend/src/api/public/v1/akrites-external/index.ts:31

• The bulk route consumes one limiter token while starting up to 20 workflows, and this same change also raises the shared limiter default from 5 to 50. At the hard batch size that permits 1,000 costly workflows/hour (and increases the existing single-job allowance 10×), so the limiter no longer bounds workflow-start cost as described. Keep the prior default unless the increase is explicitly required, and make bulk submissions consume tokens proportional to jobs.length or use a separate weighted limiter.

  max:
    Number.isSafeInteger(blastRadiusRateLimitMax) && blastRadiusRateLimitMax > 0
      ? blastRadiusRateLimitMax
      : 50,

backend/src/api/public/v1/packages/submitBlastRadiusJobBatch.ts:35

• Temporal client initialization is awaited before entering the per-job try blocks. On the first request after startup, getPackagesTemporalClient() can reject when Temporal is unreachable, causing the whole endpoint to return an error instead of the documented 202 with one failed result per job. Move/propagate client initialization into submitOneJob's guarded path while still sharing one initialization promise.

  const packagesTemporal = await getPackagesTemporalClient()

services/apps/packages_worker/src/blast-radius/workflows.ts:46

• A heartbeat timeout requires heartbeats during the activity, but blastRadiusReport only calls heartbeat() after runReportStage has completed. This makes one minute the effective timeout for the report (stricter than its two-minute start-to-close timeout) without detecting a stuck query. Remove this option or heartbeat while the report work is in progress.

  heartbeatTimeout: '1 minute',

backend/src/api/public/v1/akrites-external/openapi.yaml:1259

• The new poll route is rate-limited and can return 429, but this operation omits that response even though the single-job poll contract documents it. Add the 429 error response so generated clients and consumers have the complete contract.

                $ref: '#/components/schemas/Error'

backend/src/api/public/v1/packages/submitBlastRadiusJobBatch.ts:35

• getPackagesTemporalClient() performs Connection.connect(), so awaiting it before submitOneJob's try/catch makes a Temporal outage reject the entire request with a 500 and no per-job results. This contradicts the documented 202/failed-entry semantics. Acquire the shared client inside each per-job try (the cached promise still prevents duplicate connections) so every job can be marked and returned as failed.

  const packagesTemporal = await getPackagesTemporalClient()

backend/src/api/public/v1/packages/getBlastRadiusJobBatch.ts:51

• PostgreSQL serializes UUID values in lowercase, while z.uuid() accepts uppercase hexadecimal. An uppercase but valid requested ID therefore exists in analysisById under its lowercase DB representation and is incorrectly returned as found: false. Normalize only for the lookup while continuing to echo the original ID.

    const analysis = analysisById.get(requestedAnalysisId)

services/apps/packages_worker/src/blast-radius/dependentsScan.ts:411

• Cancellation is treated as a normal loop exit here. The scan then returns partial results, and runDependentsStage persists them and marks the stage succeeded; a retry can consequently skip the incomplete stage. Propagate cancellation with signal.throwIfAborted() rather than silently breaking, including before the scan returns.

    if (analyzed.length >= topN || signal?.aborted) break

Comments suppressed due to low confidence (3)

backend/src/api/public/v1/packages/submitBlastRadiusJobBatch.ts:35

• getPackagesTemporalClient() performs Connection.connect() and can reject when Temporal is unreachable, but this acquisition happens before the per-job try/catch. In that failure mode the whole handler rejects without returning the documented 202/failed entry per job. Move client acquisition into the guarded per-job path (after creating the pending row), or convert an acquisition failure into one failed result for every requested job.

  const packagesTemporal = await getPackagesTemporalClient()

backend/src/api/public/v1/akrites-external/openapi.yaml:462

• The response can contain status: failed when row creation or workflow start fails, so it is not true that every job is submitted. Describe this as one result per requested job to avoid promising acceptance for failed entries.

      description: >
        Plain array in request order, one entry per submitted job — unlike the
        read batches there is no found/not-found case, every job is submitted.

services/apps/packages_worker/src/blast-radius/dependentsScan.ts:98

• The concurrency values come from numeric environment variables but are not validated. With a negative value, Array.from creates zero workers and both the dependents scan and reachability stage can complete successfully without processing any items; fractional values also silently change the effective cap. Reject anything other than a positive integer, as the existing packages-worker concurrency utility does.

): Promise<void> {
  let next = 0

• Files reviewed: 21/24 changed files
• Comments generated: 0 new
• Review effort level: Medium

Comments suppressed due to low confidence (4)

backend/src/api/public/v1/akrites-external/index.ts:38

• The default was raised from 5 to 50 requests/hour while this same limiter charges a 20-job batch only once. That permits up to 1,000 paid reachability workflows per hour (and also raises the existing single-job allowance 10×), so this no longer provides the strict cost guard described for the route. Preserve the previous budget or introduce a quota weighted by the number of submitted jobs.

// Blast-radius jobs default to 50 requests/hour.
const blastRadiusRateLimiter = envTunableRateLimiter(
  'AKRITES_BLAST_RADIUS_RATE_LIMIT',
  50,
  60 * 60 * 1000,

services/apps/packages_worker/src/blast-radius/dependentsScan.ts:43

• The shared fetch itself should remain independent of one job, but a cancelled caller currently still awaits that signal-less promise. During phase 1 this keeps all in-flight workers alive until their requests settle, allowing the timed-out activity to overlap its retry—the zombie behavior this change is intended to prevent. Make each caller's wait abortable without cancelling the shared underlying fetch.

// highImpactNames' singleflight above, generalized per-key). Deliberately not tied to
// any individual caller's cancellation signal — callers pass a signal-less fetchFn, so
// one job's cancellation can't abort a fetch other jobs are still waiting on. Error

services/apps/packages_worker/src/blast-radius/dependentsScan.ts:393

• This cache key omits start and end, although both affect the downloads API result. Across a UTC date boundary, a new analysis can reuse the previous 30-day window for up to 10 minutes and rank/select dependents using stale counts. Include the requested range in the key.

      const result = await pointRangeCache(name, () =>
        withRateLimitRetry(() =>
          fetchPointRange(name, isoDate(rangeStart), isoDate(rangeEnd), undefined),
        ),
      )

backend/src/api/public/v1/akrites-external/openapi.yaml:1260

• The batch poll route is protected by rateLimiter, so it can return 429, but this new operation omits that response while the adjacent single-job poll documents it. Add the 429 response so generated clients and API consumers receive the complete contract.

        '403':
          description: Token missing read:packages scope.
          content:
            application/json:
              schema:
                $ref: '#/components/schemas/Error'

• Files reviewed: 21/24 changed files
• Comments generated: 0 new
• Review effort level: Medium

Comments suppressed due to low confidence (2)

services/apps/packages_worker/src/npm/signals.ts:11

• These { once: true } listeners are removed only if their source signal aborts, not when the fetch finishes. Because the Temporal cancellation signal is shared by every request in an activity, each completed request leaves a closure/controller retained on that signal and a scan can accumulate enough listeners to leak memory and trigger listener warnings. The worker runs on Node 20, so use native AbortSignal.any behind a narrow type assertion (or return a cleanup callback and remove both listeners in each fetch's finally).

  internal.addEventListener('abort', () => controller.abort(internal.reason), { once: true })
  external.addEventListener('abort', () => controller.abort(external.reason), { once: true })

backend/src/api/public/v1/akrites-external/index.ts:34

• The comment says the default is 50 requests/hour, but the limiter immediately below is configured with defaultMax = 5. This makes operational tuning guidance incorrect.

// Blast-radius jobs default to 50 requests/hour.

• Files reviewed: 21/24 changed files
• Comments generated: 0 new
• Review effort level: Medium